package internal

import (
	"context"
	"fmt"
	"strconv"
	"strings"
	"time"

	"gitee.com/JMArch/service_addr_syncer/pkg/delayqueue"
	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/mvcc/mvccpb"
	"github.com/zeast/logs"
)

// 同步kv到etcd，加上指定的前缀
type EtcdSyncer struct {
	cli            *clientv3.Client
	prefix         string
	serviceManager *serviceManager
	delayQueue     *delayqueue.DelayQueue
	delDelay       time.Duration

	lastEventMap map[string]*mvccpb.KeyValue // 记录service上一次的变更操作
}

type serviceManager struct {
	services []*serviceInfo
	keyMap   map[string]*serviceInfo   // map for etcd key: test-service.10.127.0.0.1:12345
	nameMap  map[string][]*serviceInfo // map for service name: test-service
}

func newServiceManager() *serviceManager {
	m := &serviceManager{
		keyMap:  make(map[string]*serviceInfo),
		nameMap: make(map[string][]*serviceInfo),
	}

	return m
}

func (m *serviceManager) initServices(services []*serviceInfo) {
	m.services = services
	for _, service := range services {
		m.keyMap[service.key] = service
		m.nameMap[service.name] = append(m.nameMap[service.name], service)
	}
	// m.printServices()
}

func (m *serviceManager) diffService(services []*serviceInfo) (adds []*serviceInfo, dels []*serviceInfo) {
	for _, serv := range services {
		if !keyInServices(serv.key, m.services) {
			adds = append(adds, serv)
		}
	}

	for _, serv := range m.services {
		if !keyInServices(serv.key, services) {
			dels = append(dels, serv)
		}
	}
	return
}

func (m *serviceManager) putService(service *serviceInfo) {
	m.services = append(m.services, service)
	m.keyMap[service.key] = service
	m.nameMap[service.name] = append(m.nameMap[service.name], service)
	// m.printServices()
}

func (m *serviceManager) delService(service *serviceInfo) {
	var index int
	for i, serv := range m.services {
		if serv.key == service.key {
			index = i
			break
		}
	}
	m.services = append(m.services[:index], m.services[index+1:]...)
	delete(m.keyMap, service.key)
	nameServices := m.nameMap[service.name]
	for i, serv := range nameServices {
		if serv.key == service.key {
			index = i
			break
		}
	}
	nameServices = append(nameServices[:index], nameServices[index+1:]...)
	m.nameMap[service.name] = nameServices
	// m.printServices()
}

func (m *serviceManager) printServices() {
	fmt.Printf("service list:\n")
	for _, serv := range m.services {
		fmt.Printf("\t%s:%s\n", serv.key, serv.version)
	}

	fmt.Printf("service key map:\n")
	for key, serv := range m.keyMap {
		fmt.Printf("\t%s -> %s:%s\n", key, serv.key, serv.version)
	}

	fmt.Printf("service name map:\n")
	for name, servs := range m.nameMap {
		fmt.Printf("\t%s:\n", name)
		for _, serv := range servs {
			fmt.Printf("\t\t%s:%s\n", serv.key, serv.version)
		}
	}
}

func keyInServices(key string, services []*serviceInfo) bool {
	for _, serv := range services {
		if key == serv.key {
			return true
		}
	}
	return false
}

type serviceInfo struct {
	key     string
	name    string
	workNum int
	addr    string
	version string
	kv      *mvccpb.KeyValue
}

// TODO: watch RpcPool.UseBackup
func NewEtcdSyncer(endPoints []string, prefix string, delDelay time.Duration) (*EtcdSyncer, error) {
	cli, err := clientv3.New(clientv3.Config{
		Endpoints:            endPoints,
		AutoSyncInterval:     time.Second,
		DialTimeout:          time.Second,
		DialKeepAliveTime:    time.Second,
		DialKeepAliveTimeout: time.Second,
	})
	if err != nil {
		return nil, err
	}
	s := &EtcdSyncer{
		cli:          cli,
		prefix:       prefix,
		delayQueue:   delayqueue.NewDelayQueue(),
		delDelay:     delDelay,
		lastEventMap: make(map[string]*mvccpb.KeyValue),
	}
	err = s.initServices()
	if err != nil {
		return nil, err
	}
	go s.delayLoop()
	return s, nil
}

// 从备份的etcd中获取当前所有的service信息
func (s *EtcdSyncer) getCurrentServices() ([]*serviceInfo, error) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()

	resp, err := s.cli.Get(ctx, s.prefix, clientv3.WithPrefix())
	if err != nil {
		return nil, err
	}

	services := s.decodeServices(resp.Kvs)
	return services, nil
}

func (s *EtcdSyncer) initServices() error {
	currentServices, err := s.getCurrentServices()
	if err != nil {
		return err
	}

	m := newServiceManager()
	m.initServices(currentServices)
	s.serviceManager = m
	return nil
}

func (s *EtcdSyncer) putService(service *serviceInfo) error {
	key := s.prefix + service.key
	val := service.version
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()
	_, err := s.cli.Put(ctx, key, val)
	return err
}

func (s *EtcdSyncer) delService(service *serviceInfo) error {
	key := s.prefix + service.key
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*3)
	defer cancel()
	_, err := s.cli.Delete(ctx, key)
	return err
}

// 同步新增数据
func (s *EtcdSyncer) syncPut(kv *mvccpb.KeyValue) {
	service, err := s.decodeService(kv)
	if err != nil {
		logs.Errorf("decode %s: %s", kv.Key, err.Error())
		return
	}
	err = s.putService(service)
	if err != nil {
		logs.Errorf("put service %s: %s", service.key, err.Error())
		return
	}
	s.serviceManager.putService(service)
	// 将本次操作记录到该service最近的一次同步操作中
	s.lastEventMap[service.key] = kv
	logs.Debugf("etcd put sync %s->%s", service.key, service.version)
}

// 同步删除数据
func (s *EtcdSyncer) syncDel(kv *mvccpb.KeyValue) {
	service, err := s.decodeService(kv)
	if err != nil {
		logs.Errorf("decode %s:%s", kv.Key, err.Error())
		return
	}
	s.delayQueue.Enqueue(service, s.delDelay)
}

// 延时处理删除的操作
func (s *EtcdSyncer) delayLoop() {
	for item := range s.delayQueue.Chan() {
		service := item.(*serviceInfo)
		lastEvent := s.lastEventMap[service.key]
		// 如果该del操作发生在上一次同步的操作之前，则该del不需要同步，因为已经不是最新的
		if lastEvent != nil && lastEvent.ModRevision > service.kv.ModRevision {
			continue
		}
		s.doSyncDel(service)
	}
}

func (s *EtcdSyncer) doSyncDel(service *serviceInfo) {
	err := s.delService(service)
	if err != nil {
		logs.Errorf("delete service %s: %s", service.key, err.Error())
		return
	}
	s.serviceManager.delService(service)
	logs.Debugf("etcd delete sync %s->%s", service.key, service.version)
}

// 初始化时，同步当前的配置到新的etcd
func (s *EtcdSyncer) syncInit(kvs []*mvccpb.KeyValue) {
	services := s.decodeServices(kvs)
	adds, dels := s.serviceManager.diffService(services)
	for _, add := range adds {
		err := s.putService(add)
		if err != nil {
			logs.Errorf("init put service %s: %s", add.key, err.Error())
			continue
		}
		s.serviceManager.putService(add)
	}

	for _, del := range dels {
		err := s.delService(del)
		if err != nil {
			logs.Errorf("init delete service %s: %s", del.key, err.Error())
			continue
		}
		s.serviceManager.delService(del)
	}
}

/*
lark.address.my-test-service.5.127.0.0.1:9090
0.2.2
*/
func (s *EtcdSyncer) decodeService(kv *mvccpb.KeyValue) (*serviceInfo, error) {
	key := string(kv.Key)
	data := strings.SplitN(key, ".", 5)
	if len(data) != 5 {
		return nil, fmt.Errorf("etcd key is wrong. %s", key)
	}
	workNum, err := strconv.Atoi(data[3])
	if err != nil {
		return nil, fmt.Errorf("parse etcd key %s:%s", key, err.Error())
	}
	info := &serviceInfo{
		key:     strings.Join(data[2:], "."),
		name:    data[2],
		workNum: workNum,
		addr:    data[4],
		version: string(kv.Value),
		kv:      kv,
	}
	return info, nil
}

func (s *EtcdSyncer) decodeServices(kvs []*mvccpb.KeyValue) []*serviceInfo {
	var services = make([]*serviceInfo, 0, len(kvs))
	for _, kv := range kvs {
		serviceInfo, err := s.decodeService(kv)
		if err != nil {
			logs.Error(err.Error())
			continue
		}
		services = append(services, serviceInfo)
	}

	return services
}
